package com.tornainbow;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.avro.AvroRuntimeException;
import org.apache.avro.Schema;
import org.apache.avro.file.CodecFactory;
import org.apache.avro.file.DataFileStream;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author mcrysler
 */
public class StreamingService {
    // Define a logger for this class
    private static final Logger logger = LoggerFactory.getLogger(StreamingService.class);
    
    // Define the schema that will be used to serialize/deserialize objects
    private static final ReflectData reflectData = ReflectData.get();
    private static final Schema schema = reflectData.getSchema(Record.class);
    
    // Define private class variables
    private DataFileWriter out;
    private DataFileStream in;
    
    /**
     * Sets the private class variables
     * @param is The input stream from which data will be deserialized
     * @param os The output stream to which data will be serialized
     */
    public StreamingService(InputStream is, OutputStream os) {
        try {
            this.in = new DataFileStream(is, new ReflectDatumReader<Record>());
        } catch (IOException e) {
            logger.error("Failed to instantiate a data file stream object", e);
        }
        
        try {
            this.out = new DataFileWriter(new ReflectDatumWriter<Record>()).
                    //setCodec(CodecFactory.deflateCodec(9)).
                    create(schema, os);
        } catch (IOException e) {
            logger.error("Failed to instantiate a data file writer object", e);
        }
    }
    
    /**
     * Sets the private class variables
     * @param is The input stream from which data will be deserialized
     */
    public StreamingService(InputStream is) {
        try {
            this.in = new DataFileStream(is, new ReflectDatumReader<Record>());
        } catch (IOException e) {
            logger.error("Failed to instantiate a data file stream object", e);
        }
    }
    
    /**
     * Sets the private class variables
     * @param os The output stream to which data will be serialized
     */
    public StreamingService(OutputStream os) {
        try {
            this.out = new DataFileWriter(new ReflectDatumWriter<Record>()).
                    setCodec(CodecFactory.deflateCodec(9)).
                    create(schema, os);
        } catch (IOException e) {
            logger.error("Failed to instantiate a data file writer object", e);
        }
    }
    
    /**
     * Serializes the data to the output stream
     * @param record The data to serialize to the output stream
     */
    public void serialize(Record record) {
        // Make sure the data file writer has been properly initialized
        if (this.out == null) {
            logger.error("Failed to serialize the data due to a null data file writer");
            return;
        }
        try {
            // Write the data to the output stream
            this.out.append(record);
            this.out.flush();
        } catch (IOException e) {
            logger.error("Failed to serialize the data", e);
        }
    }
    
    /**
     * Deserializes the data from the input stream
     * @return The data that has been deserialized from the input stream
     */
    public Record deserialize() {
        // Make sure the data file stream has been properly initialized
        if (this.in == null) {
            logger.error("Failed to deserialize the data due to a null data file stream");
            return null;
        }
        // Read the data from the input stream
        Record record = null;
        try {
            if (this.in.hasNext()) {
                record = (Record) in.next();
            }
        } catch (AvroRuntimeException e) {
            return record;
        }
        return record;
    }
}
